package com.zyx.flinkdemo.stream.utils;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.io.IOException;
import java.io.Serializable;
import java.util.Properties;

/**
 * @author zyx
 * @since 2021/04/19 22:38
 */
public class KafkaUtil implements Serializable {

    public static String getKafkaServer(String runEnv) {
        String kafkaServer;
        switch (runEnv) {
            case "dev": default:
                kafkaServer = "10.12.13.22:9092,10.12.13.23:9092,10.12.13.24:9092";
                break;
            case "test":
                kafkaServer = "10.12.13.22:9092";
                break;
            case "pro":
                kafkaServer = "";
                break;
        }
        return kafkaServer;
    }

    public static FlinkKafkaConsumer<String> getKafkaSource(String topic, String groupId, String kafkaServer) throws IOException {
        Properties prop = new Properties();
        /*Properties kafkaProp = new Properties();
        Properties runProp = new Properties();
        runProp.load(KafkaUtil.class.getClassLoader().getResourceAsStream("run-env.properties"));
        String runEnv = runProp.getProperty("run.env");
        kafkaProp.load(
                KafkaUtil.class.getClassLoader().getResourceAsStream("kafka-"+runEnv+".properties"));
        String kafkaServer = kafkaProp.getProperty("kafka.server");*/
        // prop.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, CommonConfig.KAFKA_SERVER);
        prop.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServer);
        prop.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        return new FlinkKafkaConsumer<String>(topic, new SimpleStringSchema(), prop);
    }
}
